package repeater

import (
    "../common"
    "../dmap"
    "container/list"
    "encoding/json"
    "fmt"
    "github.com/gin-gonic/gin"
    "net/http"
    "strconv"
    "strings"
    "sync"
    "time"
)

const BasePath = "golia/repeater"

//TODO: Clean unactive ip

type CacheWriter struct {
    mu               sync.RWMutex
    rc               *RepeaterController
    HightWaterOffset int64
    LowWaterOffset   int64
    WorkingOffset    int64
    datapointLimit   int
    Metrics          map[string]map[Metric]*Cache
    etcd_addrs       []string
    u, p             string
    dmap             dmap.DistributeMap
    ttl              int64
}

type Cache struct {
    Values            *list.List `json:"-"`
    MessageCount      int64      `json:"last_message_count"`
    MessageFromTime   uint32     `json:"last_message_from_time"`
    MessageToTime     uint32     `json:"last_message_from_time"`
    LastMessageTime   uint32     `json:"last_message_time"`
    LastActiveTime    uint32     `json:"last_active_time"`
    LastMessageOffset int32      `json:"last_message_time_offset"`
}

type Metric struct {
    Head string
    Sub  string
    Path string
}

func (m Metric) String() string {
    if m.Path == "" {
        return m.Head + "." + m.Sub
    }
    return m.Head + "." + m.Path + "." + m.Sub
}
func (m Metric) match(s string) bool {
    return strings.HasPrefix(m.String(), s)
}

func (s *Cache) merge(o *Cache) {
    s.MessageCount += o.MessageCount
    if s.MessageFromTime == 0 || s.MessageFromTime > o.MessageFromTime {
        s.MessageFromTime = o.MessageFromTime
    }
    if s.MessageToTime == 0 || s.MessageToTime < o.MessageToTime {
        s.MessageToTime = o.MessageToTime
    }
    if s.LastMessageTime == 0 || s.LastMessageTime < o.LastMessageTime {
        s.LastMessageTime = o.LastMessageTime
    }
    if s.LastMessageOffset == 0 || s.LastMessageOffset < o.LastMessageOffset {
        s.LastMessageOffset = o.LastMessageOffset
    }
    if s.LastActiveTime < o.LastMessageTime {
        s.LastActiveTime = o.LastMessageTime
    }
}

func (s *Cache) PromeString() string {
    dp := s.Values.Front().Value.(common.DataTuple)
    return fmt.Sprintf("%f %d000", dp.Val, dp.Time)
}

func (s *Cache) points(limit int) (ret [][2]float64) {
    for e := s.Values.Back(); e != nil; e = e.Prev() {
        dp := e.Value.(common.DataTuple)
        ret = append(ret, [2]float64{dp.Val, float64(dp.Time)})
        if len(ret) == limit {
            return
        }
    }
    return
}

func (s *Cache) String(limit int) string {
    if s.Values == nil {
        return "N/A"
    }
    var ret []string
    for e := s.Values.Back(); e != nil; e = e.Prev() {
        dp := e.Value.(common.DataTuple)
        ret = append(ret, fmt.Sprintf("(%d,%f)", dp.Time, dp.Val))
        if len(ret) == limit {
            break
        }
    }
    return strings.Join(ret, ",")
}

func (s *Cache) LastValue() (uint32, float64) {
    if s.Values != nil {
        dp := s.Values.Front().Value.(common.DataTuple)
        return dp.Time, dp.Val
    }
    return 0, -1
}

func (s *Cache) inc(dp *common.DataPoint, size int) {
    s.MessageCount++
    if s.Values == nil {
        s.Values = list.New()
    }
    s.Values.PushFront(dp.GetTuple())
    if s.Values.Len() > size {
        last := s.Values.Back()
        s.Values.Remove(last)
    }
    if s.MessageFromTime > dp.Time {
        s.MessageFromTime = dp.Time
    }
    if s.MessageToTime < dp.Time {
        s.MessageToTime = dp.Time
    }
    s.LastMessageTime = dp.Time
    s.LastActiveTime = uint32(time.Now().Unix())
    s.LastMessageOffset = int32(s.LastActiveTime - s.LastMessageTime)
}

func (cs *CacheWriter) SetController(rc *RepeaterController) error {
    cs.rc = rc
    return nil
}
func (cs *CacheWriter) Commit() (int64, error) {
    cs.mu.RLock()
    defer cs.mu.RUnlock()
    return cs.WorkingOffset, nil
}

type CacheData struct {
    Target     string       `json:"target"`
    Datapoints [][2]float64 `json:"Datapoints"`
}

func (cs *CacheWriter) Handle(dps *DataPointPack) error {
    cs.mu.Lock()
    defer cs.mu.Unlock()
    cs.WorkingOffset = dps.Offset()
    for i := 0; i < len(dps.Datapoints); i++ {
        dp := dps.Datapoints[i]
        ret := common.Extract(dp.Name)
        if len(ret) == 0 {
            continue
        }
        ip := ret["ip"]
        m := Metric{ret["head"], ret["sub"], ret["path"]}
        if _, ok := cs.Metrics[ip]; !ok {
            cs.Metrics[ip] = make(map[Metric]*Cache)
            if cs.dmap != nil {
                if err := cs.dmap.Add(ip); err != nil {
                    return err
                }
            }
        }
        if _, ok := cs.Metrics[ip][m]; !ok {
            cs.Metrics[ip][m] = &Cache{}
        }
        cs.Metrics[ip][m].inc(dp, cs.datapointLimit)
        if cs.LowWaterOffset == 0 || cs.LowWaterOffset+1 == dps.Offsets[i] {
            cs.LowWaterOffset = dps.Offsets[i]
        }
        if cs.HightWaterOffset < dps.Offsets[i] {
            cs.HightWaterOffset = dps.Offsets[i]
        }
    }
    return nil
}
func (cs *CacheWriter) Start() {
    cs.rc.wg.Add(1)
}
func (cs *CacheWriter) Stop() error {
    defer cs.rc.waitGroup().Done()
    if cs.dmap != nil {
        return cs.dmap.Close()
    }
    return nil
}
func (cs *CacheWriter) Name() string {
    return "static"
}

func (cs *CacheWriter) PromeFormat() string {
    cs.mu.RLock()
    defer cs.mu.RUnlock()
    var lines []string
    for k, v := range cs.Metrics {
        addr := "IP:" + k
        lines = append(lines, fmt.Sprintf("# TYPE %s gauge", addr))
        for metric, vv := range v {
            var line string
            if metric.Path == "" {
                line = fmt.Sprintf(`%s{head="%s",sub="%s"} %s`, addr, metric.Head, metric.Sub, vv.PromeString())
            } else {
                line = fmt.Sprintf(`%s{head="%s",sub="%s",path="%s"} %s`, addr, metric.Head, metric.Sub, metric.Path, vv.PromeString())
            }
            lines = append(lines, line)
        }
    }
    return strings.Join(lines, "\n")
}

func (cs *CacheWriter) toSummaryString(addr, prefix string) string {
    var lines []string
    if ret, ok := cs.Metrics[addr]; ok {
        for m, v := range ret {
            if !m.match(prefix) {
                continue
            }
            content, _ := json.Marshal(v)
            lines = append(lines, fmt.Sprintf("%s\t%s", m.String(), string(content)))
        }
    }
    return strings.Join(lines, "\n")
}

func (cs *CacheWriter) toString(addr, prefix string, limit int) string {
    var lines []string
    if ret, ok := cs.Metrics[addr]; ok {
        for m, v := range ret {
            if !m.match(prefix) {
                continue
            }
            lines = append(lines, fmt.Sprintf("%s\t%s", m.String(), v.String(limit)))
        }
    }
    return strings.Join(lines, "\n")
}

func (cs *CacheWriter) toJsonString(addr, prefix string, limit int) string {
    var data []CacheData
    if ret, ok := cs.Metrics[addr]; ok {
        for m, v := range ret {
            if !m.match(prefix) {
                continue
            }
            data = append(data, CacheData{Target: m.String(), Datapoints: v.points(limit)})
        }
    }
    content, _ := json.Marshal(data)
    return string(content)
}
func (cs *CacheWriter) Init() error {
    if len(cs.etcd_addrs) != 0 {
        var err error
        if cs.dmap, err = dmap.NewDistributeMap(BasePath, fmt.Sprintf("%d", cs.rc.Partition), cs.u, cs.p, cs.etcd_addrs, cs.ttl); err != nil {
            return err
        }
        if err = cs.dmap.Init(); err != nil {
            return err
        }
    }
    cs.rc.restServer.GET("/Datapoints", func(c *gin.Context) {
        var addrs []string
        cs.mu.RLock()
        defer cs.mu.RUnlock()
        for addr := range cs.Metrics {
            addrs = append(addrs, addr)
        }
        c.JSON(200, addrs)
    })
    cs.rc.restServer.GET("/Datapoints/:addr", func(c *gin.Context) {
        addr := c.Param("addr")
        cs.mu.RLock()
        defer cs.mu.RUnlock()
        s := &Cache{}
        if ret, ok := cs.Metrics[addr]; ok {
            for _, v := range ret {
                s.merge(v)
            }
        }
        c.JSON(200, s)
    })
    cs.rc.restServer.GET("/export", func(c *gin.Context) {
        c.String(http.StatusOK, cs.PromeFormat())
    })
    cs.rc.restServer.GET("/Datapoints/:addr/:metric", func(c *gin.Context) {
        addr := c.Param("addr")
        metric := c.Param("metric")
        show := c.Query("show")
        limit := 1
        if l, err := strconv.Atoi(c.Query("count")); err == nil {
            limit = l
        }
        if limit > cs.datapointLimit {
            limit = cs.datapointLimit
        }
        cs.mu.RLock()
        defer cs.mu.RUnlock()
        var ret string
        if show == "info" {
            ret = cs.toSummaryString(addr, metric)
        } else if show == "json" {
            ret = cs.toJsonString(addr, metric, limit)
        } else {
            ret = cs.toString(addr, metric, limit)
        }
        if ret != "" {
            c.String(http.StatusOK, ret)
        } else {
            c.AbortWithStatus(http.StatusNotFound)
        }
    })
    return nil
}

func NewCacheWriter(cfg map[string]string) *CacheWriter {
    l := 30
    if s, ok := cfg["datapoints_limit"]; ok {
        if i, err := strconv.Atoi(s); err != nil {
            l = i
        }
    }
    var etcd_addrs []string
    if s, ok := cfg["etcd_addrs"]; ok {
        if ss := strings.Split(s, ","); len(ss) != 0 {
            etcd_addrs = ss
        }
    }
    var u, p string
    ttl := 5
    if s, ok := cfg["ttl"]; ok {
        if i, err := strconv.Atoi(s); err != nil {
            ttl = i
        }
    }
    if s, ok := cfg["uname"]; ok {
        u = s
    }
    if s, ok := cfg["passwd"]; ok {
        p = s
    }
    if u == "" || p == "" {
        u, p = "", ""
    }
    return &CacheWriter{
        Metrics:        make(map[string]map[Metric]*Cache),
        datapointLimit: l,
        etcd_addrs:     etcd_addrs,
        u:              u,
        p:              p,
        ttl:            int64(ttl),
    }
}
